﻿// ———————————————————————–
// <copyright file="ContentProcessingDAL.cs" company="The MITRE Corporation">
//    Copyright (c) 2010 The MITRE Corporation. All rights reserved.
//    Licensed under the Apache License, Version 2.0 (the "License");
//    you may not use this file except in compliance with the License.
//    You may obtain a copy of the License at
//    http://www.apache.org/licenses/LICENSE-2.0
//    Unless required by applicable law or agreed to in writing, software
//    distributed under the License is distributed on an "AS IS" BASIS,
//    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//    See the License for the specific language governing permissions and
//    limitations under the License.
// </copyright>
// ———————————————————————–
/////////////////////////////////////////////////////////////////////////////////////////////////
// ContentProcessingDAL.cs - 
// Project: "ICNETDAL" - AWS
//
// Language:    C#, .NET 4.0
// Platform:    Windows 7, VS 2010
// Author:      Don McGarry The MITRE Corporation
/////////////////////////////////////////////////////////////////////////////////////////////////

// Copyright (c) 2010 The MITRE Corporation. All rights reserved.
//
// NOTICE
// This software was produced for the U. S. Government
// under Contract No. FA8721-09-C-0001, and is
// subject to the Rights in Noncommercial Computer Software
// and Noncommercial Computer Software Documentation Clause
// (DFARS) 252.227-7014 (JUN 1995)

using System;
using System.Collections.Generic;
using System.Globalization;
using System.Linq;
using System.Text;
using System.Xml;
using System.Xml.Linq;
using Amazon.DynamoDBv2;
using Amazon.DynamoDBv2.Model;
using Amazon.Util;
using EDXLSharp.EDXLDELib;
using GISUtil;
using GISUtil.HTML;
using ICNETServices;

namespace ICNETDAL.AWS
{
  public class DynamoContentCache : IContentCache
  {
    /// <summary>
    /// Log4net logging object
    /// </summary>
    private static readonly log4net.ILog Log = log4net.LogManager.GetLogger(System.Reflection.MethodBase.GetCurrentMethod().DeclaringType);

    /// <summary>
    /// Amazon Dynamo client
    /// </summary>
    private AmazonDynamoDBClient DynamoClient;

    public DynamoContentCache()
      : base()
    {
      AmazonDynamoDBConfig config = new AmazonDynamoDBConfig();
      config.ServiceURL = AWSConstants.DynamoServiceURL;
      DynamoClient = new AmazonDynamoDBClient(config);
    }

    /// <summary>
    /// Gets all KML
    /// </summary>
    /// <returns>All KML</returns>
    public string GetAllKML()
    {
      List<Dictionary<string, AttributeValue>> allItems = this.RunAllQuery(new List<string> {"_KML"} );

      StringBuilder sbKML = new StringBuilder();

      foreach (Dictionary<string, AttributeValue> row in allItems)
      {
        string sKML = row["_KML"].S;
        if (!string.IsNullOrWhiteSpace(sKML))
        {
          sbKML.Append(sKML);
        }
      }

      return sbKML.ToString();
    }

    /// <summary>
    /// Gets all XML
    /// </summary>
    /// <returns>All XML</returns>
    public string GetAllXML()
    {
      List<Dictionary<string, AttributeValue>> allItems = this.RunAllQuery(new List<string> { "contentObject" });

      StringBuilder sbXML = new StringBuilder();

      foreach (Dictionary<string, AttributeValue> row in allItems)
      {
        string sXML = row["contentObject"].S;
        if (!string.IsNullOrWhiteSpace(sXML))
        {
          XElement contentXML = XElement.Parse(sXML);
          IEnumerable<XElement> xmlContent = contentXML.Elements("xmlContent");
          foreach (XElement xml in xmlContent)
          {
            IEnumerable<XElement> embeddedXML = xml.Elements("embeddedXMLContent");
            foreach (XElement embedded in embeddedXML)
            {
              IEnumerable<XElement> embeddedContent = embedded.Elements();
              foreach (XElement message in embeddedContent)
              {
                string msg = message.ToString();
                sbXML.Append(msg);
              }
            }
          }

        }
      }

      return sbXML.ToString();
    }
    
    /// <summary>
    /// Gets the currently active KML
    /// </summary>
    /// <returns>Currently active KML</returns>
    public string GetActiveKML()
    {
      List<Dictionary<string, AttributeValue>> allItems = this.RunActiveQuery(new List<string> { "_KML" });

      StringBuilder sbKML = new StringBuilder();

      foreach (Dictionary<string, AttributeValue> row in allItems)
      {
        string sKML = row["_KML"].S;
        if (!string.IsNullOrWhiteSpace(sKML))
        {
          sbKML.Append(sKML);
        }
      }

      return sbKML.ToString();
    }
    
    /// <summary>
    /// Gets the currently active XML
    /// </summary>
    /// <returns>Currently active XML</returns>
    public string GetActiveXML()
    {
      List<Dictionary<string, AttributeValue>> allItems = this.RunActiveQuery(new List<string> { "contentObject" });

      StringBuilder sbXML = new StringBuilder();

      foreach (Dictionary<string, AttributeValue> row in allItems)
      {
        string sXML = row["contentObject"].S;
        if (!string.IsNullOrWhiteSpace(sXML))
        {
          XElement contentXML = XElement.Parse(sXML);
          IEnumerable<XElement> xmlContent = contentXML.Elements("xmlContent");
          foreach (XElement xml in xmlContent)
          {
            IEnumerable<XElement> embeddedXML = xml.Elements("embeddedXMLContent");
            foreach (XElement embedded in embeddedXML)
            {
              IEnumerable<XElement> embeddedContent = embedded.Elements();
              foreach (XElement message in embeddedContent)
              {
                string msg = message.ToString();
                sbXML.Append(msg);
              }
            }
          }

        }
      }

      return sbXML.ToString();
    }

    /// <summary>
    /// Gets the expired KML
    /// </summary>
    /// <returns>Expired KML</returns>
    public string GetExpiredKML()
    {
      try
      {
        StringBuilder sbKML = new StringBuilder();

        ScanResult result = null;

        do
        {
          ScanRequest req = new ScanRequest();
          req.TableName = AWSConstants.ContentProcessing.DynamoInfo.TableName;

          Condition cond = new Condition();
          cond.ComparisonOperator = ComparisonOperator.LT;
          cond.AttributeValueList = new List<AttributeValue>();

          AttributeValue val = new AttributeValue();
          val.S = DateTime.UtcNow.ToString(AWSSDKUtils.ISO8601DateFormat);
          cond.AttributeValueList.Add(val);
          req.ScanFilter.Add("_ExpiresTime", cond);

          if (result != null)
          {
            req.ExclusiveStartKey = result.LastEvaluatedKey;
          }

          result = DynamoClient.Scan(req);

          foreach (Dictionary<string, AttributeValue> row in result.Items)
          {
            string sKML = row["_KML"].S;
            if (!string.IsNullOrWhiteSpace(sKML))
            {
              sbKML.Append(sKML);
            }
          }

        } while (result.LastEvaluatedKey != null && result.LastEvaluatedKey.Count != 0);


        return sbKML.ToString();
      }
      catch (Exception e)
      {
        Log.Error("Error scanning from dynamo ContentCache for all KML", e);
        return string.Empty;
      }
    }

    /// <summary>
    /// Gets all KML that has been updated since the specified time
    /// </summary>
    /// <param name="since">The cutoff for updates</param>
    /// <returns>Updated KML since the specified time</returns>
    public string GetUpdatedKML(DateTime since)
    {
      //TODO: ICNET.AWS.ContentProcessingDAL.GetUpdatedKML
      throw new NotImplementedException();
    }

    public string GetActiveXMLByRole(string role, string roleURI)
    {
      List<Dictionary<string, AttributeValue>> allItems = this.RunActiveByRoleQuery(role, roleURI, new List<string> { "contentObject" });

      StringBuilder sbXML = new StringBuilder();

      foreach (Dictionary<string, AttributeValue> row in allItems)
      {
        string sXML = row["contentObject"].S;
        if (!string.IsNullOrWhiteSpace(sXML))
        {
          XElement contentXML = XElement.Parse(sXML);
          IEnumerable<XElement> xmlContent = contentXML.Elements("xmlContent");
          foreach (XElement xml in xmlContent)
          {
            IEnumerable<XElement> embeddedXML = xml.Elements("embeddedXMLContent");
            foreach (XElement embedded in embeddedXML)
            {
              IEnumerable<XElement> embeddedContent = embedded.Elements();
              foreach (XElement message in embeddedContent)
              {
                string msg = message.ToString();
                sbXML.Append(msg);
              }
            }
          }

        }
      }

      return sbXML.ToString();
    }

    public string GetActiveKMLByRole(string role, string roleURI)
    {
      List<Dictionary<string, AttributeValue>> allItems = this.RunActiveByRoleQuery(role, roleURI, new List<string> { "_KML" });

      StringBuilder sbKML = new StringBuilder();

      foreach (Dictionary<string, AttributeValue> row in allItems)
      {
        string sKML = row["_KML"].S;
        if (!string.IsNullOrWhiteSpace(sKML))
        {
          sbKML.Append(sKML);
        }
      }

      return sbKML.ToString();
    }

    public string GetExpiredKMLByRole(string role, string roleURI)
    {
      //TODO: ICNET.AWS.ContentProcessingDAL.GetExpiredKMLByRole
      throw new NotImplementedException();
    }

    public string GetUpdatedKMLByRole(DateTime since, string role, string roleURI)
    {
      //TODO: ICNET.AWS.ContentProcessingDAL.GetUpdatedKMLByRole
      throw new NotImplementedException();
    }

    /// <summary>
    /// Gets the XML representation of a ContentObject.
    /// </summary>
    /// <param name="contentHash">The hash of the content object to retrieve.</param>
    /// <returns>String of XML content to return to the user.</returns>
    public string GetContentObjectByContentHash(int contentHash)
    {
      return GetFieldByHash(contentHash.ToString(), "contentObject");
    }

    /// <summary>
    /// Gets the KML representation of a ContentObject.
    /// </summary>
    /// <param name="contentHash">The hash of the content object to retrieve.</param>
    /// <returns>String of KML content to return to the user.</returns>
    public string GetKMLByContentHash(int contentHash)
    {
      return GetFieldByHash(contentHash.ToString(), "_KML");
    }

    /// <summary>
    /// Gets the GeoRSS representation of a ContentObject.
    /// </summary>
    /// <param name="contentHash">The hash of the content object to retrieve.</param>
    /// <returns>String of GeoRSS content to return to the user.</returns>
    public string GetGeoRSSByContentHash(int contentHash)
    {
      return GetFieldByHash(contentHash.ToString(), "_GeoRSS");
    }

    /// <summary>
    /// Gets the HTML representation of a ContentObject.
    /// </summary>
    /// <param name="contentHash">The hash of the content object to retrieve.</param>
    /// <returns>String of HTML content to return to the user.</returns>
    public string GetHTMLByContentHash(int contentHash)
    {
      return GetFieldByHash(contentHash.ToString(), "_HTML");      
    }
    
    /// <summary>
    /// Deletes cached content with the provided Sender ID and Distribution ID
    /// </summary>
    /// <param name="senderId">The DE SenderID</param>
    /// <param name="distributionId">The DE DistributionID</param>
    public void DeleteContent(string senderId, string distributionId)
    {
      QueryRequest itemstodelete = new QueryRequest()
      {
        TableName = AWSConstants.ContentProcessing.DynamoInfo.TableName,
        IndexName = "distributionID-senderID-index" // TODO: make this a constant in the configs
      };

      Dictionary<String, Condition> keyConditions = new Dictionary<string, Condition>();

      keyConditions.Add(
        "distributionID",
        new Condition()
        {
          ComparisonOperator = "EQ",
          AttributeValueList = { new AttributeValue { S = distributionId } }
        });

      keyConditions.Add(
        "senderID",
        new Condition()
        {
          ComparisonOperator = "EQ",
          AttributeValueList = { new AttributeValue { S = senderId } }
        });

      itemstodelete.KeyConditions = keyConditions;

      QueryResponse itemstodeleteresp = DynamoClient.Query(itemstodelete);
      List<Dictionary<string, AttributeValue>> result = itemstodeleteresp.Items;
      foreach (Dictionary<string, AttributeValue> dbitem in result)
      {
        DeleteItemRequest delreq = new DeleteItemRequest();
        delreq.TableName = AWSConstants.ContentProcessing.DynamoInfo.TableName;
        delreq.Key = new Dictionary<string, AttributeValue> { { "_ContentHash", new AttributeValue { N = dbitem["_ContentHash"].N } } };

        try
        {
          DeleteItemResponse delresp = DynamoClient.DeleteItem(delreq);
          if (delresp.HttpStatusCode == System.Net.HttpStatusCode.OK)
          {
            Log.Info("Item deleted from dynamo");
          }
          else
          {
            Log.Error("Error deleting from dynamo" + delresp.HttpStatusCode.ToString());
          }
        }
        catch (Exception e)
        {
          Log.Error("Error deleting from dynamo", e);
        }
      }
    }

    /// <summary>
    /// Deletes cached content with the provided Content Hash
    /// </summary>
    /// <param name="contentHash">The content hash</param>
    public void DeleteContent(string contentHash)
    {
      DeleteItemRequest delreq = new DeleteItemRequest();
      delreq.TableName = AWSConstants.ContentProcessing.DynamoInfo.TableName;
      delreq.Key = new Dictionary<string, AttributeValue> { { "_ContentHash", new AttributeValue { N = contentHash } } };

      try
      {
        DeleteItemResponse delresp = DynamoClient.DeleteItem(delreq);
        if (delresp.HttpStatusCode == System.Net.HttpStatusCode.OK)
        {
          Log.Info("Item deleted from dynamo");
        }
        else
        {
          Log.Error("Error deleting from dynamo" + delresp.HttpStatusCode.ToString());
        }
      }
      catch (Exception e)
      {
        Log.Error("Error deleting from dynamo", e);
      }
    }

    /// <summary>
    /// Adds content objects to the content cache, or updates existing entries
    /// </summary>
    /// <param name="de">DE containing the content objects</param>
    public void AddAndUpdateContent(EDXLDE de)
    {
      //add new, update existing, remove others
      List<Dictionary<string, AttributeValue>> toUpdate = GetCurrentContentHashes(de.DistributionID, de.SenderID);

      if(toUpdate.Any(ci => IsNewer(ci, de.DateTimeSent)))
      {
        //We have a newer version of this message already
        Log.Info("Found content from same or more recent DE; discarding this message.");
        return;
      }

      //Add/update the content objects we got from this DE, and remove them from the toUpdate list
      foreach (ContentObject co in de.ContentObjects)
      {
        PutItemRequest putRequest = new PutItemRequest();
        putRequest.TableName = AWSConstants.ContentProcessing.DynamoInfo.TableName;

        Dictionary<string, AttributeValue> putItem = new Dictionary<string, AttributeValue>();

        string contentHash = DEDalUtils.ComputeHash(de.DistributionID, de.SenderID, co.ContentDescription).ToString();

        AttributeValue dynamovalue = new AttributeValue();
        dynamovalue.N = contentHash;
        putItem.Add("_ContentHash", dynamovalue);

        dynamovalue = new AttributeValue();
        dynamovalue.S = de.DistributionID;
        putItem.Add("distributionID", dynamovalue);

        dynamovalue = new AttributeValue();
        dynamovalue.S = de.SenderID;
        putItem.Add("senderID", dynamovalue);

        dynamovalue = new AttributeValue();
        dynamovalue.S = de.DateTimeSent.ToUniversalTime().ToString(AWSSDKUtils.ISO8601DateFormat);
        putItem.Add("dateTimeSent", dynamovalue);

        dynamovalue = new AttributeValue();
        DateTime expiration = co.ExpiresTime.HasValue ? co.ExpiresTime.Value : de.DateTimeSent.AddDays(1);
        dynamovalue.S = expiration.ToString(AWSSDKUtils.ISO8601DateFormat);
        putItem.Add("_ExpiresTime", dynamovalue);

        // write the xml for just the contentObject
        StringBuilder coBody = new StringBuilder();
        XmlWriter coWriter = XmlWriter.Create(coBody, new XmlWriterSettings()
        {
          OmitXmlDeclaration = true
        });
        co.WriteXML(coWriter);
        coWriter.Flush();
        coWriter.Close();

        dynamovalue = new AttributeValue();
        dynamovalue.S = coBody.ToString();
        putItem.Add("contentObject", dynamovalue);

        dynamovalue = new AttributeValue();
        string dcsHash = DEDalUtils.ComputeHash(de.DistributionID, de.SenderID, co.ContentDescription).ToString();
        string sKML = GISConverter.CreateKMLForContentObject(de, co, dcsHash);
        dynamovalue.S = sKML;
        putItem.Add("_KML", dynamovalue);

        //TODO: Create GeoRSS String
        //dynamovalue = new AttributeValue();
        //string sGeoRSS = AWS_GeoRSSDal.CreateGeoRSSForContentObject(de, co);
        //dynamovalue.S = sGeoRSS;
        //dynamoitem.Add("_GeoRSS", dynamovalue);

        dynamovalue = new AttributeValue();
        dynamovalue.S = HTMLEngine.GenerateHTMLForContentObject(co);
        putItem.Add("_HTML", dynamovalue);

        putRequest.Item = putItem;

        try
        {
          PutItemResponse putResponse = DynamoClient.PutItem(putRequest);
          if (putResponse.HttpStatusCode == System.Net.HttpStatusCode.OK)
          {
            Log.Info("Message Added to Dynamo");
          }
          else
          {
            Log.Error("Dynamo Error" + putResponse.HttpStatusCode.ToString());
          }
        }
        catch (Exception e)
        {
          Log.Error("Error writing DE to dynamo", e);
        }

        // remove this item from the list of things to update since we updated it
        toUpdate.RemoveAll(tu => tu["_ContentHash"].N == contentHash);
      }

      //Anything left in the toUpdate list exists in dynamo, but was not
      //in the latest DE, and should be deleted
      foreach (Dictionary<string, AttributeValue> item in toUpdate)
      {
        DeleteContent(item["_ContentHash"].N);
      }
    }

    /// <summary>
    /// Gets the value of a specified field from the cache
    /// for an entry that matches the provided hash, or null
    /// if no match to the hash or unknown field
    /// </summary>
    /// <param name="hash">The hash to lookup</param>
    /// <param name="field">The field to look up</param>
    /// <returns>Field matching the given hash, or null</returns>
    private string GetFieldByHash(string hash, string field)
    {
      List<string> valid = new List<string>() { "contentObject", "_KML", "_GeoRSS", "_HTML" };
      if (!valid.Contains(field))
      {
        return null;
      }

      QueryRequest req = new QueryRequest(AWSConstants.ContentProcessing.DynamoInfo.TableName)
      {
        AttributesToGet = new List<string> { field },
        KeyConditions = new Dictionary<string, Condition>
        {
          { "_ContentHash", new Condition() 
            { ComparisonOperator = ComparisonOperator.EQ, AttributeValueList = new List<AttributeValue> { new AttributeValue() { N = hash }} } 
          }
        }
      };

      try
      {
        QueryResponse resp = DynamoClient.Query(req);

        Dictionary<string, AttributeValue> item = resp.Items.SingleOrDefault();
        if (item != null && item.ContainsKey(field))
        {
          return item[field].S;
        }
        return null;
      }
      catch (Exception e)
      {
        Log.Error("Error scanning from dynamo ContentCache for " + field + " for content hash " + hash, e);
        throw;
      }
    }

    /// <summary>
    /// Queries the distributionID-senderID-index GSI for all content that is in Dynamo from this distributionID and senderID.
    /// This returns the list of items from Dynamo that have the distributionID, senderID, _ContentHash and dateTimeSent for 
    /// each item.
    /// </summary>
    /// <param name="distributionID">The distributionID to search for.</param>
    /// <param name="senderID">The senderID to search for.</param>
    /// <returns>List of content items for the given DE with just their _ContentHash and dateTimeSent.</returns>
    private List<Dictionary<string, AttributeValue>> GetCurrentContentHashes(string distributionID, string senderID)
    {
      QueryRequest currHashesReq = new QueryRequest()
      {
        TableName = AWSConstants.ContentProcessing.DynamoInfo.TableName,
        IndexName = "distributionID-senderID-index", // TODO: make this a constant in the configs
        KeyConditions = new Dictionary<string, Condition>()
      };

      currHashesReq.KeyConditions.Add(
        "distributionID",
        new Condition()
        {
          ComparisonOperator = "EQ",
          AttributeValueList = { new AttributeValue { S = distributionID } }
        });

      currHashesReq.KeyConditions.Add(
        "senderID",
        new Condition()
        {
          ComparisonOperator = "EQ",
          AttributeValueList = { new AttributeValue { S = senderID } }
        });

      QueryResponse currHashesResp = DynamoClient.Query(currHashesReq);

      return currHashesResp.Items;
    }

    /// <summary>
    /// Checks whether a cached item is more recent than a given time
    /// </summary>
    /// <param name="cachedItem">The cached item</param>
    /// <param name="threshold">The date to compare to</param>
    /// <returns>True if the cached item is newer than the given time</returns>
    private bool IsNewer(Dictionary<string, AttributeValue> cachedItem, DateTime threshold)
    {
      DateTime itemDateTime = DateTime.ParseExact(cachedItem["dateTimeSent"].S, AWSSDKUtils.ISO8601DateFormat, new DateTimeFormatInfo());
      Log.DebugFormat("Comparing existing {0} >=? {1}", itemDateTime, threshold);
      return itemDateTime >= threshold;
    }

    private string BuildDELinkItem(Dictionary<string, AttributeValue> item)
    {
      XElement deKey = new XElement("DELink");

      deKey.Add(new XElement("distributionID", item["distributionID"].S));
      deKey.Add(new XElement("senderID", item["senderID"].S));
      deKey.Add(new XElement("dateTimeSent", DateTime.ParseExact(item["dateTimeSent"].S, AWSSDKUtils.ISO8601DateFormat, new DateTimeFormatInfo()).ToLocalTime().ToString("yyyy-MM-ddThh:mm:sszzz")));

      XElement contentKey = new XElement("contentObject");
      contentKey.Add(new XElement("Link", ICNETSOAConstants.ICNETdatalinkToHTML + item["_ContentHash"].N));
      deKey.Add(contentKey);

      return deKey.ToString();
    }

    public string GetAllDELink()
    {
      List<Dictionary<string, AttributeValue>> allItems = this.RunAllQuery(new List<string> { "_ContentHash", "distributionID", "senderID", "dateTimeSent" });
      StringBuilder sbDELink = new StringBuilder();

      foreach (Dictionary<string, AttributeValue> item in allItems)
      {
        sbDELink.Append(BuildDELinkItem(item));
      }

      return sbDELink.ToString();
    }

    public string GetActiveDELink()
    {
      List<Dictionary<string, AttributeValue>> allItems = this.RunActiveQuery(new List<string> { "_ContentHash", "distributionID", "senderID", "dateTimeSent" });
      StringBuilder sbDELink = new StringBuilder();

      foreach (Dictionary<string, AttributeValue> item in allItems)
      {
        sbDELink.Append(BuildDELinkItem(item));
      }

      return sbDELink.ToString();
    }

    public string GetExpiredDELink()
    {
      throw new NotImplementedException();
    }

    public string GetUpdatedDELink(DateTime since)
    {
      throw new NotImplementedException();
    }

    public string GetActiveDELinkByRole(string role, string roleURI)
    {
      List<Dictionary<string, AttributeValue>> allItems = this.RunActiveByRoleQuery(role, roleURI, new List<string> { "_ContentHash", "distributionID", "senderID", "dateTimeSent" });
      StringBuilder sbDELink = new StringBuilder();

      foreach (Dictionary<string, AttributeValue> item in allItems)
      {
        sbDELink.Append(BuildDELinkItem(item));
      }

      return sbDELink.ToString();
    }

    public string GetExpiredDELinkByRole(string role, string roleURI)
    {
      throw new NotImplementedException();
    }

    public string GetUpdatedDELinkByRole(DateTime since, string role, string roleURI)
    {
      throw new NotImplementedException();
    }

    private List<Dictionary<string, AttributeValue>> RunAllQuery(List<string> attributesToGet)
    {
      List<Dictionary<string, AttributeValue>> allItems = new List<Dictionary<string, AttributeValue>>();

      try
      {
        ScanResult result = null;

        do
        {
          ScanRequest req = new ScanRequest();
          req.TableName = AWSConstants.ContentProcessing.DynamoInfo.TableName;
          req.AttributesToGet = attributesToGet;

          if (result != null)
          {
            req.ExclusiveStartKey = result.LastEvaluatedKey;
          }

          result = DynamoClient.Scan(req);

          if (result.Items.Count > 0)
          {
            allItems.AddRange(result.Items);
          }


        } while (result.LastEvaluatedKey != null && result.LastEvaluatedKey.Count != 0);

      }
      catch (Exception e)
      {
        Log.Error("Error scanning from dynamo ContentCache for all items", e);
      }


      return allItems;
    }


    private List<Dictionary<string, AttributeValue>> RunActiveQuery(List<string> attributesToGet)
    {
      List<Dictionary<string, AttributeValue>> activeItems = new List<Dictionary<string, AttributeValue>>();

      try
      {
        ScanResult result = null;

        do
        {
          ScanRequest req = new ScanRequest();
          req.TableName = AWSConstants.ContentProcessing.DynamoInfo.TableName;
          req.AttributesToGet = attributesToGet;

          Condition cond = new Condition();
          cond.ComparisonOperator = ComparisonOperator.GT;
          cond.AttributeValueList = new List<AttributeValue>();

          AttributeValue val = new AttributeValue();
          val.S = DateTime.UtcNow.ToString(AWSSDKUtils.ISO8601DateFormat);
          cond.AttributeValueList.Add(val);
          req.ScanFilter.Add("_ExpiresTime", cond);

          if (result != null)
          {
            req.ExclusiveStartKey = result.LastEvaluatedKey;
          }

          result = DynamoClient.Scan(req);

          if (result.Items.Count > 0)
          {
            activeItems.AddRange(result.Items);
          }

        } while (result.LastEvaluatedKey != null && result.LastEvaluatedKey.Count != 0);


      }
      catch (Exception e)
      {
        Log.Error("Error scanning from dynamo ContentCache for all KML", e);
      }

      return activeItems;
    }

    private List<Dictionary<string, AttributeValue>> RunActiveByRoleQuery(string role, string roleURI, List<string> attributesToGet)
    {
      string feedHash = roleURI + "::" + role;

      List<Dictionary<string, AttributeValue>> allItems = new List<Dictionary<string, AttributeValue>>();


      List<Dictionary<string, AttributeValue>> contentHashesToGet = new List<Dictionary<string, AttributeValue>>();

      try
      {
        StringBuilder sbKML = new StringBuilder();

        Dictionary<string, AttributeValue> lastKeyEvaluated = null;
        do
        {
          ScanRequest req = new ScanRequest();
          req.TableName = AWSConstants.ContentFeedProcessing.DynamoInfo.TableName;

          req.ScanFilter = new Dictionary<string, Condition>()
          {
            {
              "_FeedHash",
              new Condition()
              {
                ComparisonOperator = ComparisonOperator.EQ,
                AttributeValueList = new List<AttributeValue>()
                {
                  new AttributeValue { S = feedHash }
                }
              }
            },
            {
              "_ExpiresTime",
              new Condition()
              {
                ComparisonOperator = ComparisonOperator.GT,
                AttributeValueList = new List<AttributeValue>()
                {
                  new AttributeValue { S = DateTime.UtcNow.ToString(AWSSDKUtils.ISO8601DateFormat) }
                } 
              }
            }
          };

          // for pagination if we are not on the first page or last page we dont need to add it
          if (lastKeyEvaluated != null)
          {
            req.ExclusiveStartKey = lastKeyEvaluated;
          }

          var result = DynamoClient.Scan(req);
          lastKeyEvaluated = result.LastEvaluatedKey;

          foreach (Dictionary<string, AttributeValue> row in result.Items)
          {

            contentHashesToGet.Add(new Dictionary<string, AttributeValue>()
              {
                { "_ContentHash", row["_ContentHash"] }
              });
          }
        } while (lastKeyEvaluated != null && lastKeyEvaluated.Count > 0);

        if (contentHashesToGet.Count == 0)
        {
          Log.InfoFormat("No content found for feed {0}::{1}", roleURI, role);
          return allItems;
        }

        // using a batch get request to get all content at once
        BatchGetItemRequest contentReq = new BatchGetItemRequest()
        {
          RequestItems = new Dictionary<string, KeysAndAttributes>()
          {
            {
              AWSConstants.ContentProcessing.DynamoInfo.TableName,
              new KeysAndAttributes()
              {
                Keys = contentHashesToGet,
                AttributesToGet = attributesToGet              
              }
            }
          }
        };

        BatchGetItemResult contentResp = DynamoClient.BatchGetItem(contentReq);

        if (contentResp.Responses[AWSConstants.ContentProcessing.DynamoInfo.TableName].Count() > 0)
        {
          allItems.AddRange(contentResp.Responses[AWSConstants.ContentProcessing.DynamoInfo.TableName]);
        }

        return allItems;
      }
      catch (Exception e)
      {
        Log.Error("Error scanning from dynamo ContentCache for all KML", e);
        return allItems;
      }
    }
  }
}